package org.example.realtime.jtp.ods.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;


/**
 * @Title: CdcDateEtlMapFunction
 * @Author Lianzy
 * @Package org.example.realtime.jtp.ods.function
 * @Date 2025/6/2 16:42
 * @description 将Flink CDC实时同步的数据提取核心字段值并且转换值最后拼接为JSON字符串返回
 */
public class CdcDateEtlMapFunction extends RichMapFunction<String,String> {

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
    }

    @Override
    public String map(String value) throws Exception {
        //传入的JSON类型的字符串 先解析
        JSONObject jsonObject = JSON.parseObject(value);

        //字段after值
        String after = jsonObject.getString("after");
         //字段before
        String before = jsonObject.getString("before");
         //字段op
         String op = jsonObject.getString("op");
         //字段ts_ms
         Long ts_ms = jsonObject.getLong("ts_ms");
         //字段db 和table
        JSONObject object = jsonObject.getJSONObject("source");
        String db = object.getString("db");
        String tablevalue = object.getString("table");

        String operatorValue;
        switch (op) {
            case "u":
                operatorValue = "update";
                break;
            case "r":
            case "c":
                operatorValue = "insert";
                break;
            case "d":
                operatorValue = "delete";
                break;
            default:
                operatorValue = "unknown";
        }

        //拼接JSON字符串
        JSONObject newObject = new JSONObject();
        newObject.put("db_name",db);
        newObject.put("table_name",tablevalue);
        newObject.put("operate_type",operatorValue);
        newObject.put("ts_ms",ts_ms);
        //由于data值为json字符串{}需要先解析JSONobject对象在设置值
        if("d".equals(op)){
            newObject.put("operate_date",JSON.parseObject(before));
        }else{
             newObject.put("operate_date",JSON.parseObject(after));
        }

        return newObject.toJSONString();
    }

    @Override
    public void close() throws Exception {
        super.close();
    }
}
